package sync

import (
	"sync"
	"sync/atomic"
	"time"

	"gddgame.cc/galaxy/asteroid"
	"gddgame.cc/galaxy/asteroid/cluster"
	"gddgame.cc/galaxy/binocle"
	"gddgame.cc/galaxy/common/feature/orbit"
	"gddgame.cc/galaxy/common/feature/orbit/sync_protocol"
	"gddgame.cc/galaxy/satellite/formula/comet"
	"gddgame.cc/galaxy/utils"
	"gddgame.cc/galaxy/utils/serialize"

	"gddgame.cc/galaxy/utils/def"
)

type BlockDelegate interface {
	Join(member *memberInstance, node cluster.Node)
	Leave(member *memberInstance, node cluster.Node)
	UniqueId() uint32
	Data(...interface{})
	Location(openId string, location []byte)
}
type BlockService struct {
	def.Logger
	binocle.Service

	GetSocketClient func(socket *comet.Socket) *Client
	// 确保MaxDelay和MinDelay之间有一定间隔，避免对象在网络状态中不停跳动
	MaxDelay int    // 最大可接收延迟
	MinDelay int    // 高可用最小延迟
	AvailGap uint32 // 最低可接收阀值
}

func (service *BlockService) ID() string {
	return "syncBlock"
}

func (service *BlockService) InitService(node cluster.Node, instance *blockInstance, delegate BlockDelegate) {
	ticker := time.NewTicker(time.Minute * 1)
	go func() {
		ctx := node.Context()
		for {
			select {
			case <-ctx.Done():
				ticker.Stop()
				return
			case <-ticker.C:
				t := time.Now().Unix()
				if !instance.connecting || node == nil {
					break
				}
				for i := 0; i < len(instance.Times); i++ {
					tt := instance.Times[i]
					if tt.Update(t, instance, node) {
						if i == 0 {
							instance.Times = instance.Times[1:]
						} else if i == len(instance.Times)-1 {
							instance.Times = instance.Times[0:i]
						} else {
							instance.Times = append(instance.Times[:i], instance.Times[i+1:]...)
						}
						i -= 1
					}
				}
				if instance.autoAgent {
					node.Broadcast().Send("ping")
				}
			}
		}
	}()
	node.On("pong", func(message asteroid.Message) {
		if !instance.autoAgent {
			return
		}

		b := serialize.GetBag()
		message.UnSerialize(b)
		openId := b.ReadString(0)
		delay := b.ReadInt(1)
		b.Release()

		member, ok := instance.Members[openId]
		if !ok {
			return
		}
		//fmt.Println("block ping start")
		// 网速可用，或者可用人数不足阀值
		if delay != 0 && delay < service.MinDelay || instance.AvailNum*100/instance.MemberNum < service.AvailGap {
			// 有效delay
			instance.setAvail(openId, member, node)
		} else if delay > service.MaxDelay && instance.AvailNum > 2 {
			// 无效delay
			instance.resetAvail(openId, member, node, false)
		}
		member.Delay = delay
		//fmt.Println("block ping end")
	})
	node.On(cluster.JoinEvent, func(message asteroid.Message) {
		id := message.GetOrigin()
		other := message.GetFrom()
		openId := utils.ToString(message.Data())
		if _, ok := instance.Members[openId]; ok {
			message.Response(false)
			return
		}
		message.Response(true)
		node.Debugf("Sync Join:%s <- %s", openId, other)
		member := &memberInstance{
			OpenId:   openId,
			Owner:    make(map[uint32]bool),
			Agent:    sync.Map{},
			Observer: make(map[uint32]bool),
			Node:     other,
			Delay:    0,
		}
		delegate.Join(member, node)
		instance.Members[openId] = member
		atomic.AddUint32(&instance.MemberNum, 1)
		instance.calculate()
		instance.setAvail(openId, member, node)
		// 重新链接， 转发所有item的Instance消息
		for hash, _ := range instance.Items {
			item := instance.Items[hash]
			b := serialize.GetBag()
			service.InstanceBag(hash, item, b)
			node.Broadcast().Node(other).TargetId(id).Data(b).Send(sync_protocol.Instance)
			b.Release()

			// 持久化消息
			if item.Message != nil {
				msg := serialize.GetBag()
				msg.Append(hash, sync_protocol.MessageQueue, item.Version, item.Message)
				node.Broadcast().Node(other).TargetId(id).Data(msg).Send(sync_protocol.Queue)
				msg.Release()
			}
			// 将属于该用户的item返回
			if item.Owner == openId {
				instance.SetAgent(hash, member)
			}
		}
		// 发送长期执行
		for index := range instance.Times {
			instance.Times[index].Send(node)
		}
		//fmt.Println("Sync Join end")
	})
	node.On(cluster.LeaveEvent, func(message asteroid.Message) {
		openId := utils.ToString(message.Data())
		node.Debugf("Sync Leave:%s", openId)
		member, ok := instance.Members[openId]
		if !ok {
			message.Response(false)
			return
		}
		delegate.Leave(member, node)
		delete(instance.Members, openId)
		atomic.StoreUint32(&instance.MemberNum, instance.MemberNum-1)
		instance.calculate()
		instance.resetAvail(openId, member, node, true)
		message.Response(true)
		//fmt.Println("Sync Leave end")
	})
	node.On(cluster.CloseEvent, func(message asteroid.Message) {
		_ = node.Close()
	})
	node.On(sync_protocol.Instance, func(message asteroid.Message) {
		// id := message.Target
		b := serialize.GetBag()
		message.UnSerialize(b)
		node.Debugf("Server Instance:%#v", b.Data)
		// localid byte, location []byte, level byte, meta []interface{}, info []interface{}, data []interface{}, ownerId string, belong string, agent string
		// 保留本地id，用于映射
		localId := b.ReadByte(0)
		// 创建唯一hash：int64
		hash := delegate.UniqueId()

		meta := b.ReadList(3)
		info := b.ReadList(4)
		data := b.ReadList(5)
		//node.Debugf("Sync Instance:%s -> %s, meta:%s, info:%s, data:%s", localId, hash, meta, info, data)
		item := &itemInstance{
			Location: b.ReadBytes(1),
			Level:    b.ReadByte(2),
			Meta:     meta,
			Info:     info,
			Data:     data,
			DataExt:  map[uint16]interface{}{},
			Owner:    b.ReadString(6),
			Belong:   b.ReadUint32(7),
			Version:  0,
			Agent:    b.ReadString(8),
			Observer: "",

			changeData: true,
		}
		location := orbit.LocationCM.DecodeLocation(item.Location)
		// 判断移动方向
		item.Direction = instance.Block.DirectionInline(location)
		instance.AddItem(hash, item, node)
		//fmt.Println("send instance", hash)

		observer := instance.FindObserver()
		instance.SetObserver(hash, observer)
		//instance.AssignObserve(node, observer, []uint32{hash})
		//fmt.Println("Instance", hash, item.Agent, item.Observer)

		service.InstanceBag(hash, item, b)
		b.Write(11, localId)
		node.Broadcast().Data(b).Send(sync_protocol.Instance)
		b.Release()
		//fmt.Println("Server Instance end")
	})
	node.On(sync_protocol.Destroy, func(message asteroid.Message) {
		id := message.GetOrigin()
		b := serialize.GetBag()
		message.UnSerialize(b)
		// hash string
		hash := b.ReadUint32(0)
		item, ok := instance.Items[hash]
		node.Debugf("Sync Destroy:%s <- %s find", hash, id)
		if !ok {
			b.Release()
			return
		}
		instance.RemoveItem(hash, item)
		//node.Debugf("Sync Destroy:%s <- %s clean", hash, id)
		node.Broadcast().OriginId(id).Data(b).Send(sync_protocol.Destroy)
		b.Release()
		//fmt.Println("Server Destroy end")
	})

	node.On("transfer", func(message asteroid.Message) {
		b := serialize.GetBag()
		message.UnSerialize(b)
		node.Debugf("Server transfer:%#v", b.Data)
		// hash int64, nodeId string
		hash := b.ReadUint32(0)
		item, ok := instance.Items[hash]
		//node.Debugf("Sync Destroy:%s <- %s find", hash, id)
		if !ok {
			b.Release()
			return
		}
		nodeId := b.ReadString(1)
		// 进行转移
		bb := serialize.GetBag()
		service.InstanceBag(hash, item, bb)
		item.transfer = instance.Service.GetWithNodeId(node.Container(), nodeId)
		item.transfer.Emit().Data(bb).CallbackTo(func(success bool, data interface{}) {
			instance.RemoveItem(hash, item)
		}).Send("transferred")
		b.Release()
		//fmt.Println("Server transfer end")
	})
	// 其他节点转移的item
	node.On("transferred", func(message asteroid.Message) {
		// id := message.Target
		b := serialize.GetBag()
		message.UnSerialize(b)
		// hash int64, location []byte, level byte, meta []interface{}, info []interface{}, data []interface{}, ownerId string, belong string, version int, assign string, localId int, indexs []interface{}, datas []interface{}
		hash := b.ReadUint32(0)

		meta := b.ReadList(3)
		info := b.ReadList(4)
		data := b.ReadList(5)
		var dataExt map[uint16]interface{}
		if !b.IsNil(11) {
			indexs := b.ReadList(11)
			datas := b.ReadList(12)
			dataExt = make(map[uint16]interface{}, len(indexs))
			for key, index := range indexs {
				dataExt[utils.ToUInt16(index)] = datas[key]
			}
		} else {
			dataExt = map[uint16]interface{}{}
		}

		//_ = b.Read(6, &dataExt)
		//node.Debugf("Sync Instance:%s -> %s, meta:%s, info:%s, data:%s", localId, hash, meta, info, data)
		item := &itemInstance{
			Location: b.ReadBytes(1),
			Level:    b.ReadByte(2),
			Meta:     meta,
			Info:     info,
			Data:     data,
			DataExt:  dataExt,
			Owner:    b.ReadString(6),
			Belong:   b.ReadUint32(7),
			Version:  b.ReadByte(8),
			Agent:    b.ReadString(9),
			Observer: "",

			changeData: true,
		}
		location := orbit.LocationCM.DecodeLocation(item.Location)
		// 判断移动方向
		item.Direction = instance.Block.DirectionInline(location)
		instance.AddItem(hash, item, node)
		observer := instance.FindObserver()
		instance.SetObserver(hash, observer)
		instance.AssignObserve(node, observer, []uint32{hash})

		service.InstanceBag(hash, item, b)
		// 发送给所有用户
		node.Broadcast().Data(b).Send(sync_protocol.Instance)
		b.Release()
		message.Response(true)
	})
	node.On(sync_protocol.Queue, func(message asteroid.Message) {
		b := serialize.GetBag()
		message.UnSerialize(b)
		//fmt.Println("queue", b.Data)
		// hash int64, type byte, localVersion, xx
		// Info: [location []byte, level byte, content []interface{}]
		// Message: [code byte, data []byte, ping byte]
		// Data: [index[]byte, data []interface{}]
		hash := b.ReadUint32(0)
		item, ok := instance.Items[hash]
		if !ok {
			b.Release()
			return
		}
		if item.transfer != nil {
			// 在转移中，发送给新
			item.transfer.Emit().Data(b).Send(sync_protocol.Queue)
			b.Release()
			return
		}
		t := b.ReadByte(1)
		// 版本控制
		item.Version = b.ReadByte(2)

		index := 2
		// Info
		if t&sync_protocol.InfoQueue > 0 && !b.IsNil(index+1) {
			index++
			info := b.ReadBag(index)
			item.Location = info.ReadBytes(0)
			location := orbit.LocationCM.DecodeLocation(item.Location)
			// 判断移动方向
			direction := instance.Block.DirectionInline(location)
			orbit.LocationCM.ReleaseLocation(location)
			if direction != item.Direction {
				service.InstanceBag(hash, item, b)
				// 发送给所有用户
				node.Broadcast().Data(b).Send(sync_protocol.Instance)
				item.Direction = direction
			}
			//node.Debugf("Sync Info:%s <- %s", hash, b.Data)
			item.Level = info.ReadByte(1)
			if !info.IsNil(2) {
				content := info.ReadInterface(2)
				contentI := content.([]interface{})
				for key := range contentI {
					//fmt.Println("info", key, item.Info, infoI[key])
					if contentI[key] != nil {
						item.Info[key] = contentI[key]
					}
				}
			}
			info.Release()
		}

		// Message
		if t&sync_protocol.MessageQueue > 0 && !b.IsNil(index+1) {
			index++
			mes := b.ReadBag(index)
			// 根据类型处理
			t := mes.ReadByte(0)
			if t&sync_protocol.StandingMessage > 0 {
				// 持久化
				item.Message = mes.Bytes()
			} else if t&sync_protocol.CancelMessage > 0 {
				// 取消持久化
				item.Message = nil
			}
			mes.Release()
		}

		// Data
		if t&sync_protocol.DataQueue > 0 && !b.IsNil(index+1) {
			index++
			data := b.ReadBag(index)
			indexs := data.ReadList(0)
			datas := data.ReadList(1)
			l := uint16(len(item.Data))
			for key, index := range indexs {
				i := utils.ToUInt16(index)
				if i >= l {
					if datas[key] == nil {
						delete(item.DataExt, i)
					} else {
						item.DataExt[i] = datas[key]
					}
					item.changeData = true
				} else {
					item.Data[utils.ToByte(index)] = datas[key]
				}
			}
			// 发送同步消息，对应item
			delegate.Data(hash, indexs, datas)
			data.Release()
		}
		b.Release()
		message.Response(item.Version)
		//fmt.Println("queue end")
	})
	node.On(sync_protocol.Apply, func(message asteroid.Message) {
		id := message.GetOrigin()
		b := serialize.GetBag()
		message.UnSerialize(b)
		// hash string, type byte, wantId string, openId string
		hash := b.ReadUint32(0)
		applyType := b.ReadByte(1)
		item, ok := instance.Items[hash]
		if !ok {
			b.Release()
			return
		}
		switch sync_protocol.ApplyType(applyType) {
		case sync_protocol.OwnerApplyType:
			wantId := b.ReadString(2)
			openId := b.ReadString(3)
			if item.Owner == "" {
				item.Owner = wantId
			} else if item.Agent == openId {
				item.Owner = wantId
			} else {
				b.Release()
				return
			}
		case sync_protocol.BelongApplyType:
			belong := b.ReadUint32(2)
			openId := b.ReadString(3)
			if item.Belong == 0 {
				item.Belong = belong
			} else if item.Agent == openId {
				item.Belong = belong
			} else {
				b.Release()
				return
			}
		}
		service.Debugf("[ Apply ] %#v %#v -> %#v", applyType, hash, item)
		node.Broadcast().OriginId(id).Data(b).Send(sync_protocol.Apply)
		b.Release()
	})
	node.On(sync_protocol.Time, func(message asteroid.Message) {
		b := serialize.GetBag()
		message.UnSerialize(b)
		// code byte location []byte [info]
		// trigger: [time int, code byte, payload []]
		// move: [startTime int, endTime int, payload [], position [], index int]
		instance.Times = append(instance.Times, NewTime(instance, b, node))
		b.Release()
	})
}
func (service *BlockService) InitClient(client *cluster.Client) {
	client.Filter("ping", func(message asteroid.Message, member def.Member) {
		//fmt.Println("filter ping start")
		socket := member.(*comet.Socket)
		delay, last := socket.Delay()
		b := serialize.GetBag()
		cc := service.GetSocketClient(socket)
		// 1分钟前的忽略
		if last.Before(time.Now().Add(-1 * time.Minute)) {
			// 如果长时间没有更新
			b.Set(cc.OpenId, 0)
		} else {
			b.Set(cc.OpenId, *delay/time.Millisecond)
		}
		client.Emit().Origin(member).Data(b).Send("pong")
		b.Release()
		//fmt.Println("filter ping end")
	})
	client.Filter(sync_protocol.Instance, func(message asteroid.Message, member def.Member) {
		//fmt.Println("filter instance start")
		socket := member.(*comet.Socket)
		b := serialize.GetBag()
		message.UnSerialize(b)
		cc := service.GetSocketClient(socket)
		if cc != nil {
			hash := b.ReadUint32(0)
			agent := b.ReadString(9)
			if cc.Service.NeedBlock() {
				location := orbit.LocationCM.DecodeLocation(b.ReadBytes(1))
				if !cc.NearBlock[0].Contain(location) {
					goto Release
				}
			}
			if _, ok := cc.BlockMap.Load(hash); !ok {
				if agent == cc.OpenId {
					cc.BlockMap.Store(hash, client)
					cc.AgentMap.Store(hash, nil)
				} else {
					observe := b.ReadString(10)
					if observe == cc.OpenId {
						cc.BlockMap.Store(hash, client)
					}
				}
				cc.WriteInstance(b.Get())
			}
		}
	Release:
		b.Release()
		//fmt.Println("filter instance end")
	})
	client.Filter(sync_protocol.Destroy, func(message asteroid.Message, member def.Member) {
		//fmt.Println("filter destroy start")
		socket := member.(*comet.Socket)
		b := serialize.GetBag()
		message.UnSerialize(b)
		cc := service.GetSocketClient(socket)
		socket.Debugf("Accept Client Destroy")
		if cc != nil {
			hash := b.ReadUint32(0)
			if _, ok := cc.BlockMap.Load(hash); ok {
				cc.BlockMap.Delete(hash)
				//fmt.Println("destroy", b.Data)
				if value, ok := cc.AgentMap.Load(hash); ok {
					if value != nil {
						orbit.LocationCM.ReleaseBlock(value.(*orbit.BlockLocation))
					}
					cc.AgentMap.Delete(hash)
				}
				cc.WriteDestroy(b.Get())
			}
		}
		b.Release()
		//fmt.Println("filter destroy end")
	})
	client.Filter(sync_protocol.Assign, func(message asteroid.Message, member def.Member) {
		socket := member.(*comet.Socket)
		b := serialize.GetBag()
		message.UnSerialize(b)
		cc := service.GetSocketClient(socket)
		if cc != nil {
			// type byte, hash []uint32, openId string
			assignType := b.ReadByte(0)
			var list []uint32
			_ = b.Read(1, &list)
			openId := b.ReadString(2)
			for k, _ := range list {
				hash := utils.ToUint32(list[k])

				switch sync_protocol.AssignType(assignType) {
				case sync_protocol.AgentAssignType:
					//fmt.Println("accept Assign Agent", openId)
					// 设置代理权限
					if openId == cc.OpenId {
						cc.BlockMap.Store(hash, client)
						cc.AgentMap.Store(hash, nil)
					} else if _, ok := cc.AgentMap.Load(hash); ok {
						cc.AgentMap.Store(hash, nil)
					}
				case sync_protocol.ObserveAssignType:
					//fmt.Println("accept Assign Observe", openId)
					// 设置观察权限
					if openId == cc.OpenId {
						cc.BlockMap.Store(hash, client)
					} else if _, ok := cc.AgentMap.Load(hash); !ok {
						cc.BlockMap.Delete(hash)
					}
				}
			}
		}
		socket.Send(sync_protocol.Assign, b.Bytes())
		b.Release()
	})
	client.Filter(sync_protocol.Queue, func(message asteroid.Message, member def.Member) {
		//fmt.Println("filter queue")
		socket := member.(*comet.Socket)
		cc := service.GetSocketClient(socket)
		if cc != nil {
			b := serialize.GetBag()
			message.UnSerialize(b)
			hash := b.ReadUint32(0)
			t := b.ReadByte(1)

			if _, ok := cc.BlockMap.Load(hash); ok {
				//fmt.Println("filter queue 1")
				// 只有客户端标记的才下发到客户端
				if t&sync_protocol.DataQueue > 0 {
					if _, ok := cc.DataMap.Load(hash); ok {
						cc.DataMap.Delete(hash)
					} else {
						// 删除Data部分并切换t
						b.Pop()
						b.Write(1, t^sync_protocol.DataQueue)
					}
				}

				//fmt.Println("filter queue 2")
				if _, ok := cc.AgentMap.Load(hash); ok {
					cc.WriteQueue(b.Get())
				} else if t == sync_protocol.InfoQueue {
					var rate int8 = 0
					if value, ok := cc.ItemRateLimit.Load(hash); ok {
						rate = value.(int8)
					}
					if rate <= 0 {
						info := b.ReadBag(3)
						tmpLocation := info.ReadBytes(0)
						info.Release()
						// info 频次控制
						rate = cc.InfoRate(tmpLocation)
					}
					if rate == 0 {
						cc.WriteQueue(b.Get())
						cc.ItemRateLimit.Delete(hash)
					} else if rate > 0 {
						rate -= 1
						cc.ItemRateLimit.Store(hash, rate)
					}
				} else {
					cc.WriteQueue(b.Get())
				}
			}
			//fmt.Println("filter queue 3")
			b.Release()
		}
		//fmt.Println("filter queue end")
	})
	client.Filter(asteroid.DefaultFilter, func(message asteroid.Message, member def.Member) {
		socket := member.(*comet.Socket)
		d, _ := message.Payload()
		socket.Send(comet.DE(utils.ToByte(message.GetEvent())), d)
	})
}

func (service *BlockService) InstanceBag(hash uint32, item *itemInstance, b *serialize.Mpb) {
	// localId 用于区分是否是本地创建，用于查找本地GameObject
	if item.changeData {
		item.indexs = make([]interface{}, len(item.DataExt))
		item.datas = make([]interface{}, len(item.DataExt))
		index := 0
		for key := range item.DataExt {
			item.indexs[index] = key
			item.datas[index] = item.DataExt[key]
			index += 1
		}
	}
	b.Set(hash, item.Location, item.Level, item.Meta, item.Info, item.Data, item.Owner, item.Belong, item.Version, item.Agent, item.Observer, 0, item.indexs, item.datas)
}

func (service *BlockService) CloseInstance(node cluster.Node, instance *blockInstance) {
	instance.Close()
}
